package com.corpgovernment.organization.mq;

import com.corpgovernment.common.apollo.OrganizationApollo;
import com.corpgovernment.common.base.BaseUserInfo;
import com.corpgovernment.common.mq.enums.MessageBizTypeEnum;
import com.corpgovernment.common.mq.producer.CommonMQProducer;
import com.corpgovernment.organization.entity.SyncEmployeeRequestEntity;
import com.corpgovernment.organization.entity.SyncOrganizationRequestEntity;
import com.corpgovernment.organization.entity.db.MbImportExcelRecordDetailLog;
import com.corpgovernment.organization.entity.db.MbOrgInfo;
import com.corpgovernment.organization.enumm.ImportExcelRecordDetailLogErrorMessageEnum;
import com.corpgovernment.organization.enumm.ServiceErrorEnum;
import com.corpgovernment.organization.service.IOrganizationInfoService;
import com.corpgovernment.organization.service.ISyncEmployeeService;
import com.corpgovernment.organization.service.ISyncOrganizationService;
import com.ctrip.corp.obt.generic.utils.CollectionUtils;
import com.ctrip.corp.obt.generic.utils.JsonUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@Component
public class MultiTenantSyncEmpAndOrgProducer {

    @Autowired
    private CommonMQProducer commonMQProducer;

    @Autowired
    private OrganizationApollo organizationApollo;
    @Autowired
    private ISyncOrganizationService syncOrganizationService;
    @Autowired
    private ISyncEmployeeService syncEmployeeService;
    @Autowired
    @Qualifier(value = "defaultThreadPoolExecutor")
    private ThreadPoolExecutor defaultThreadPoolExecutor;
    @Autowired
    private IOrganizationInfoService organizationInfoService;

    /**
     * 同步组织架构消息
     *
     * @param organizationInfo
     * @param notifyUrl
     * @param inValidList
     * @param recordId
     */
    public void sendSyncOrgMsg(SyncOrganizationRequestEntity.SyncOrganizationInfoEntity organizationInfo, String notifyUrl, MessageBizTypeEnum messageBizType, BaseUserInfo baseUserInfo, List<MbImportExcelRecordDetailLog> inValidList, Long recordId) throws InterruptedException, RemotingException, MQClientException, MQBrokerException , Exception{
        checkMsgParams(organizationInfo, inValidList, messageBizType);
        log.info("organizationInfo：{}、notifyUrl：{}、messageBizType：{}", organizationInfo, notifyUrl, messageBizType);
        SyncOrganizationRequestEntity entity = new SyncOrganizationRequestEntity();
        BeanUtils.copyProperties(baseUserInfo, entity);
        entity.setNotifyUrl(notifyUrl);
        entity.setOrgInfoList(new ArrayList<>());
        entity.getOrgInfoList().add(organizationInfo);
        this.syncOrg(messageBizType, entity);
        log.info("organizationInfo：{}", organizationInfo);
    }

    /**
     * 同步组织架构消息
     *
     * @param organizationInfo
     * @param notifyUrl
     */
    public void sendSyncOrgMsgV1(SyncOrganizationRequestEntity.SyncOrganizationInfoEntity organizationInfo, String notifyUrl, MessageBizTypeEnum messageBizType, BaseUserInfo baseUserInfo) throws Exception{
        log.info("organizationInfo：{}、notifyUrl：{}、messageBizType：{}", organizationInfo, notifyUrl, messageBizType);
        SyncOrganizationRequestEntity entity = new SyncOrganizationRequestEntity();
        BeanUtils.copyProperties(baseUserInfo, entity);
        entity.setNotifyUrl(notifyUrl);
        entity.setOrgInfoList(new ArrayList<>());
        entity.getOrgInfoList().add(organizationInfo);
        this.syncOrg(messageBizType, entity);
        log.info("organizationInfo：{}", organizationInfo);
    }

    private void checkMsgParams(SyncOrganizationRequestEntity.SyncOrganizationInfoEntity organizationInfo, List<MbImportExcelRecordDetailLog> inValidList, MessageBizTypeEnum messageBizType) {
        if (!MessageBizTypeEnum.SYNC_ORGANIZATION_DELETE.equals(messageBizType)){
            return;
        }
        // 组织架构参数校验
        if ( organizationInfo.getBasicInfo() == null || organizationInfo.getBasicInfo().getBusinessCode() == null) {
            buildInvalidInfo(organizationInfo, inValidList);
        }
        List<MbOrgInfo> mbOrgInfoList =
                organizationInfoService.listByBusinessCode(Collections.singletonList(organizationInfo.getBasicInfo().getBusinessCode()));
        if (CollectionUtils.isEmpty(mbOrgInfoList)) {
            buildInvalidInfo(organizationInfo, inValidList);
        }
    }

    private  void buildInvalidInfo(SyncOrganizationRequestEntity.SyncOrganizationInfoEntity organizationInfo, List<MbImportExcelRecordDetailLog> inValidList) {
        ServiceErrorEnum serviceErrorEnum = ServiceErrorEnum.SyncHumanResources_NewError;
        serviceErrorEnum.setErrorMessage(ImportExcelRecordDetailLogErrorMessageEnum.SyncHumanResources_SyncOrgInfo_Business_Code_Is_Empty.getErrorMessage());
        MbImportExcelRecordDetailLog mbImportExcelRecordDetailLog = new MbImportExcelRecordDetailLog();
        mbImportExcelRecordDetailLog.setLineNo(organizationInfo.getLineNo());
        mbImportExcelRecordDetailLog.setMainLineNo(organizationInfo.getLineNo());
        mbImportExcelRecordDetailLog.setStatus("F");
        mbImportExcelRecordDetailLog.setMessage(ImportExcelRecordDetailLogErrorMessageEnum.A.getErrorMessage());
        mbImportExcelRecordDetailLog.setDatachangeCreatetime(new Date());
        mbImportExcelRecordDetailLog.setDatachangeLasttime(new Date());
        inValidList.add(mbImportExcelRecordDetailLog);
    }


    private void syncOrgByThread(MessageBizTypeEnum messageBizType, SyncOrganizationRequestEntity entity) {
        String tags = messageBizType.getTags();
        defaultThreadPoolExecutor.execute(() -> {
            try {
                syncOrganizationService.syncOrgInfo(entity, tags);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    private void syncOrg(MessageBizTypeEnum messageBizType, SyncOrganizationRequestEntity entity) throws Exception{
        String tags = messageBizType.getTags();
        try {
            syncOrganizationService.syncOrgInfo(entity, tags);
        } catch (Exception e) {
            log.info("同步失败：", e);
            throw e;
        }

    }

    private SendResult syncOrgByMq(MessageBizTypeEnum messageBizType, SyncOrganizationRequestEntity entity, String orgId) throws InterruptedException {
        String msg = JsonUtils.toJsonString(entity);
        Message message = new Message(messageBizType.getTopic(), messageBizType.getTags(), msg.getBytes(StandardCharsets.UTF_8));
        SendResult sendResult = null;
        try {
            sendResult = commonMQProducer.send(message, this.orgMessageQueueSelector(), orgId);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return sendResult;
    }

    /**
     * 相同父级别的组织分到同一个queue（保证顺序消费）
     *
     * @return
     */
    private MessageQueueSelector orgMessageQueueSelector() {
        return (List<MessageQueue> mqs, Message msg, Object arg) -> mqs.get(0);
    }

    /**
     * 同步人员消息
     */
    public void sendSyncEmployeeMsg(SyncEmployeeRequestEntity employeeInfo) {

//		log.info("employeeInfo：{}、notifyUrl：{}、messageBizType：{}", employee, notifyUrl, messageBizType);
//		SyncEmployeeRequestEntity entity = new SyncEmployeeRequestEntity();
//		entity.setNotifyUrl(notifyUrl);
//		entity.setEmployeeInfoList(new ArrayList<>());
//		entity.getEmployeeInfoList().add(employee);
        String syncOrgEmployeeModel = organizationApollo.getSyncOrgEmployeeModel();
        log.info("同步组织员工模式：{}", syncOrgEmployeeModel);
        if (Objects.equals(syncOrgEmployeeModel, "MQ")) {
            this.syncEmployeeByMq(employeeInfo);
        } else {
            this.syncEmployeeByThread(employeeInfo);
        }
    }

    private void syncEmployeeByThread(SyncEmployeeRequestEntity employeeInfo) {
        defaultThreadPoolExecutor.execute(() -> {
            try {
                syncEmployeeService.syncEmployeeInfo(employeeInfo);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    private SendResult syncEmployeeByMq(SyncEmployeeRequestEntity entity) {
        SendResult sendResult = null;
        try {
            String msg = JsonUtils.toJsonString(entity);
            sendResult = commonMQProducer.send(new Message(MessageBizTypeEnum.SYNC_EMPLOYEE_NEW.getTopic(), MessageBizTypeEnum.SYNC_EMPLOYEE_NEW.getTags(), msg.getBytes(StandardCharsets.UTF_8)));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return sendResult;
    }

}
